package com.shujia.compute

import java.util.Properties

import com.google.gson.Gson
import com.shujia.bean.{CarWide, YDBean}
import com.shujia.sink.{HbaseMapper, HbaseSink}
import com.shujia.tf.YongDuWindow
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.hadoop.hbase.client.Put

object YongDu {

  def main(args: Array[String]): Unit = {

    /**
      *
      * 道路实时拥堵情况
      *
      * 判断拥堵条件
      * 1、平均速度
      * 2、车辆数量大于100
      *
      * 窗口范围（窗口大小5分钟，滑动时间1分钟）
      *
      *
      * 1、绿色
      * 车辆数量小于100
      * 平均速度45以上
      *
      * 2、黄色
      * 车辆数量大于100
      * 平均速度40-45
      *
      * 3、红色
      * 车辆数量大于100
      * 平均速度35-40
      *
      * 4、深红
      * 车辆数量大于100
      * 平均速度小于35
      *
      *
      * 输出结果
      * 道路,卡口，时间，拥堵情况
      *
      */


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置事件时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val kafkaProps = new Properties()
    kafkaProps.setProperty("zookeeper.connect", "node1:2181")
    kafkaProps.setProperty("bootstrap.servers", "node1:9092")
    kafkaProps.setProperty("group.id", "qweqwew")

    val kafkaConsumer = new FlinkKafkaConsumer[String]("dwd_car_wide", new SimpleStringSchema, kafkaProps)
    //从头读取数据
    kafkaConsumer.setStartFromEarliest()
    val lineDS = env.addSource(kafkaConsumer)

    val carWideDS = lineDS.map(line => {
      val gson = new Gson()
      gson.fromJson(line, classOf[CarWide])
    })


    //设置时间时间列和水位线
    val waterMarkDS = carWideDS.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[CarWide](Time.seconds(5)) {
      override def extractTimestamp(t: CarWide): Long = {
        t.getTs * 1000
        //指定时间戳列
      }
    })


    //窗口大小5分钟，滑动时间一分钟
    val ydDS = waterMarkDS.keyBy(car => car.getDlName + "-" + car.getKcId)
      .timeWindow(Time.minutes(5), Time.minutes(1))
      .process(new YongDuWindow)

    ydDS.print()

    val hbaseMapper = new HbaseMapper[YDBean] {
      override def mapper(value: YDBean): Put = {
        val rowKey = value.getDl + "-" + value.getKc
        val put = new Put(rowKey.getBytes())
        put.add("info".getBytes(), "state".getBytes(), value.getTs, value.getState.getBytes())

        put
      }
    }

    ydDS.addSink(new HbaseSink[YDBean]("dl_yd", hbaseMapper, "node1:2181"))

    env.execute("yd")

  }

}
